package com.ndp.fb.rdb.mongo.impl;

import java.util.ArrayList;
import java.util.List;

import com.mongodb.util.JSON;
import com.ndp.fb.rdb.model.AdGroup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import com.alibaba.dubbo.config.annotation.Service;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.mongodb.AggregationOptions;
import com.mongodb.BasicDBObject;
import com.mongodb.Cursor;
import com.mongodb.DB;
import com.mongodb.DBCollection;
import com.mongodb.DBCursor;
import com.mongodb.DBObject;
import com.mongodb.WriteConcern;
import com.ndp.fb.constant.MongoConstants;
import com.ndp.fb.mongo.api.FeedMongoService;
import com.ndp.fb.mongo.model.Feed;

@Service
public class FeedMongoServiceImpl implements FeedMongoService {
	Logger logger = LoggerFactory.getLogger(getClass());
	@Autowired
	private DB mongoDBClientPrimary;

	@Override
	public void save(List<Feed> feeds) {
		if (feeds == null) {
			return;
		}
		DBCollection dbCollection = mongoDBClientPrimary.getCollection(MongoConstants.TABLE_NAME_FEED);
		List<DBObject> o = (List<DBObject>) com.mongodb.util.JSON.parse(JSONArray.toJSONString(feeds));
		for (Feed feed : feeds) {
			dbCollection.update(new BasicDBObject("feedId", feed.getFeedId()).append("id", feed.getId()).append("updateDate", feed.getUpdateDate()),
					(DBObject) com.mongodb.util.JSON.parse(JSONObject.toJSONString(feed)), true, false,
					WriteConcern.JOURNALED);
		}
	}

    @Override
    public void saveNew(List<Feed> feedList) {
        DBCollection dbCollection = mongoDBClientPrimary.getCollection(MongoConstants.TABLE_NAME_NEW_FEED);
        List<DBObject> data = new ArrayList<>();
        for (Feed feed : feedList) {
            String jsonString = JSONObject.toJSONString(feed);
            DBObject o = (DBObject) com.mongodb.util.JSON.parse(jsonString);
            data.add(o);
        }
        dbCollection.insert(data);
    }

    @Override
	public JSONArray find(JSONObject query) {
		DBCollection dbCollection = mongoDBClientPrimary.getCollection(MongoConstants.TABLE_NAME_FEED);
		DBObject q = (DBObject) com.mongodb.util.JSON.parse(JSONObject.toJSONString(query));
		DBCursor dbCursor = dbCollection.find(q);
		JSONArray result = new JSONArray();
		if (dbCursor == null) {
			return null;
		}
		DBObject obj;
		while (dbCursor.hasNext()) {
			obj = dbCursor.next();
			result.add(obj.get("id"));
		}
		return result;
	}

	@Override
	public List<Feed> sample(int n) {
		List<DBObject> pipeline = new ArrayList<>();
		DBObject group = new BasicDBObject("size", n);
		pipeline.add(new BasicDBObject("$sample", group));
		long start = System.currentTimeMillis();
		AggregationOptions aggregationOptions = AggregationOptions.builder().allowDiskUse(true).build();
		Cursor cursor = mongoDBClientPrimary.getCollection(MongoConstants.TABLE_NAME_FEED).aggregate(pipeline,
				aggregationOptions);
		logger.info("Mongo聚合查询花费时间：" + (System.currentTimeMillis() - start) + "毫秒");
        List<Feed> feeds = collectFeed(cursor);
		return feeds;
	}

	@Override
	public List<Feed> sampleByFeed(String s, int i) {
		JSONObject queryFeed = new JSONObject();
		queryFeed.put("feedId", s);

        List<DBObject> pipeline = new ArrayList<>();
        DBObject group = new BasicDBObject("size", i);
        pipeline.add(new BasicDBObject("$match", queryFeed));
        pipeline.add(new BasicDBObject("$sample", group));
        long start = System.currentTimeMillis();
        AggregationOptions aggregationOptions = AggregationOptions.builder().allowDiskUse(true).build();
        Cursor cursor = mongoDBClientPrimary.getCollection(MongoConstants.TABLE_NAME_FEED).aggregate(pipeline,
                aggregationOptions);
        logger.info("sampleByFeed Mongo聚合查询花费时间：" + (System.currentTimeMillis() - start) + "毫秒");
        List<Feed> feeds = collectFeed(cursor);
        return feeds;
	}

    @Override
    public void deleteNewFeed(Long id, Long materialId) {
        BasicDBObject query = new BasicDBObject();
        query.put("feedId", id.toString());
        query.put("materielId", materialId.toString());
        mongoDBClientPrimary.getCollection(MongoConstants.TABLE_NAME_NEW_FEED).remove(query);
    }

    @Override
    public List<Feed> sampleByNewFeed(Long feedId, Long MaterielId, int i) {

        JSONObject queryFeed = new JSONObject();
        queryFeed.put("feedId", feedId.toString());
        queryFeed.put("materielId", MaterielId.toString());

        List<DBObject> pipeline = new ArrayList<>();
        DBObject group = new BasicDBObject("size", i);
        pipeline.add(new BasicDBObject("$match", queryFeed));
        pipeline.add(new BasicDBObject("$sample", group));
        long start = System.currentTimeMillis();
        AggregationOptions aggregationOptions = AggregationOptions.builder().allowDiskUse(true).build();
        Cursor cursor = mongoDBClientPrimary.getCollection(MongoConstants.TABLE_NAME_NEW_FEED).aggregate(pipeline,
                aggregationOptions);
        logger.info("sampleByFeed Mongo聚合查询花费时间：" + (System.currentTimeMillis() - start) + "毫秒");
        List<Feed> feeds = collectFeed(cursor);

        return feeds;
    }

    private List<Feed> collectFeed(Cursor cursor) {
        List<Feed> feeds = new ArrayList<>();
        try {
            while (cursor.hasNext()) {
                DBObject dbObject = cursor.next();
                feeds.add(JSONObject.parseObject(JSONObject.toJSONString(dbObject), Feed.class));
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            cursor.close();
        }
        return feeds;
    }

    @Override
	public List<Feed> find(JSONObject query, int skip, int limit) {
		DBCollection dbCollection = mongoDBClientPrimary.getCollection(MongoConstants.TABLE_NAME_FEED);
		DBObject q = (DBObject) com.mongodb.util.JSON.parse(JSONObject.toJSONString(query));
		DBCursor dbCursor = dbCollection.find(q).skip(skip).limit(limit);
		List<Feed> result = new ArrayList<>();
		if (dbCursor == null) {
			return null;
		}
		DBObject obj;
		while (dbCursor.hasNext()) {
			obj = dbCursor.next();
			result.add(JSONObject.parseObject(JSONObject.toJSONString(obj), Feed.class));
		}
		return result;
	}

	@Override
	public void deleteFeed(String feedId, String updateDate) {
		DBCollection dbCollection = mongoDBClientPrimary.getCollection(MongoConstants.TABLE_NAME_FEED);
		DBObject q = new BasicDBObject();
		q.put("feedId", feedId);
		q.put("updateDate", updateDate);
		dbCollection.remove(q, WriteConcern.JOURNALED);
	}

}
